package org.dh2580.delay.queue.bean.task;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.dh2580.delay.queue.bean.AbstractMsgListener;
import org.dh2580.delay.queue.bean.bean.DelayQueueBean;
import org.dh2580.delay.queue.bean.db.DBUtil;
import org.dh2580.delay.queue.bean.executor.SimpleExecutor;
import org.dh2580.delay.queue.bean.model.DelayMsg;

/**
 * @Author: hao.deng
 * @Date: 2019/5/15-上午10:46
 */
public class SelectExpiredMsgTask extends SelectMsgTask {

    private static final String QUERY_EXPIRE_SQL_TPL = "SELECT id AS id, route_key AS routeKey, msg AS msg, ttl AS ttl, ack_state AS ackState, create_time AS createTime FROM tb_delay_msg_queue WHERE ttl > ? AND ttl < ? AND ack_state = 1;";

    private static final String QUERY_NEW_SQL_TPL = "SELECT id AS id, route_key AS routeKey, msg AS msg, ttl AS ttl, ack_state AS ackState, create_time AS createTime FROM tb_delay_msg_queue WHERE ttl > ? AND ttl < ? AND ack_state = 0;";

    private static final String BATCH_INIT_SQL_TPL = "UPDATE tb_delay_msg_queue SET ack_state = 0 WHERE id IN (?) AND ack_state = 1";

    private Integer maxExpiredMinute;

    public SelectExpiredMsgTask(SimpleExecutor listenerExecutor, SimpleExecutor callbackExecutor, Map<String, List<AbstractMsgListener>> listeners, Integer maxExpiredMinute) {
        super(listenerExecutor, callbackExecutor, listeners, maxExpiredMinute);

        this.maxExpiredMinute = maxExpiredMinute;
    }


    @Override
    public List<DelayMsg> selectDelayMsgs() throws Exception {
        long cur = System.currentTimeMillis();

        long ttl = cur - TimeUnit.MINUTES.toMillis(maxExpiredMinute);
        long minTtl = cur - TimeUnit.DAYS.toMillis(DelayQueueBean.ABANDON_THRESHOLD_DAY);

        List<DelayMsg> newDelayMsgs = DBUtil.query(QUERY_NEW_SQL_TPL, DelayMsg.class, minTtl, ttl);

        List<DelayMsg> expiredDelayMsgs = DBUtil.query(QUERY_EXPIRE_SQL_TPL, DelayMsg.class, minTtl, ttl);

        StringBuilder msgIdStr = new StringBuilder();
        for (DelayMsg delayMsg : expiredDelayMsgs) {
            msgIdStr.append(delayMsg.getId());
            msgIdStr.append(",");
        }

        if (msgIdStr.length() > 0) {
            int affectRows = DBUtil.execute(BATCH_INIT_SQL_TPL.replace("?", msgIdStr.subSequence(0, msgIdStr.length() - 1)));
            if (affectRows > 0) {
                newDelayMsgs.addAll(expiredDelayMsgs);
            }
        }

        return newDelayMsgs;
    }
}
